package com.ksc.wordcount.rpc.Executor;

import akka.actor.ActorRef;
import akka.actor.ActorSystem;
import com.ksc.wordcount.driver.DriverEnv;
import com.ksc.wordcount.worker.ExecutorEnv;
import com.typesafe.config.Config;
import com.typesafe.config.ConfigFactory;

import java.time.Duration;
import java.util.Arrays;
import java.util.HashMap;
import java.util.Map;

/**
 * 该类为ExecutorSystem，用于创建和管理Executor的ActorSystem实例。
 *
 * 注释：
 *
 * akka.actor.provider：定义使用远程Actor系统。
 * akka.remote.transport：指定使用Netty进行远程传输。
 * akka.remote.netty.tcp.hostname：指定主机名。
 * akka.remote.netty.tcp.port：指定端口号。
 * getExecutorSystem()：获取Executor的ActorSystem实例，如果已存在则直接返回；否则根据配置创建一个新的ActorSystem实例。
 * getDriverRef()：获取Driver的ActorRef，根据Driver的URL，使用actorSelection方法获取ActorRef，并等待最多10秒钟进行解析。
 */
public class ExecutorSystem {

    static  ActorSystem system; // 静态变量，用于保存ActorSystem实例

    public static ActorSystem getExecutorSystem() {
        if(system!=null){ // 如果ActorSystem实例已存在，直接返回
            return system;
        }
        Map<String, Object> map=new HashMap<>();

        map.put("akka.actor.provider","remote"); // 设置使用远程Actor系统
        map.put("akka.remote.transport","akka.remote.netty.NettyRemoteTransport"); // 设置使用Netty传输
        map.put("akka.remote.netty.tcp.hostname", ExecutorEnv.host); // 设置主机名
        map.put("akka.remote.netty.tcp.port", ExecutorEnv.port); // 设置端口号
        Config config = ConfigFactory.parseMap(map).withFallback(ConfigFactory.load()); // 使用配置Map创建配置实例
        system = ActorSystem.create("ExecutorSystem", config); // 创建ActorSystem实例
        return system;
    }

    public static ActorRef getDriverRef(){
        String driverUrl = ExecutorEnv.driverUrl; // 获取Driver的URL
        ActorRef driverRef = getExecutorSystem().actorSelection(driverUrl) // 根据URL获取Driver的ActorRef
                .resolveOne(Duration.ofSeconds(10)).toCompletableFuture().join(); // 解析ActorRef，等待最多10秒钟
        return driverRef;
    }
}
